package my.thesis.analytics.metrics;

import java.util.Date;

import org.joda.time.DateTime;

import backtype.storm.metric.api.IMetric;

public class ThroughputMetric implements IMetric
{
	DateTime startTime;
	DateTime currentTime;
	DateTime endTime;
	long tuplesProcessed;
	
	long tupleAck;
	
	//KALEITAI STHN PREPARE METHODO TWN PERSISTORS META TO INITIALIZATION TWN METRICS
	public void setStartTime()
	{
		this.startTime=DateTime.now();
		this.tuplesProcessed=0;
		this.tupleAck=0;
	}
		
	//KALEITAI STHN EXECUTE METHODO TWN PERSISTORS
	public void incrTuplesBy(long incrementBy) 
	{ 
		tuplesProcessed += incrementBy;
	} 
	
	//KALEITAI STHN EXECUTE METHODO TWN PERSISTORS
	public void incrTuples() 
	{ 
		tuplesProcessed ++;
	}
	
	public void incrTuplesAck() 
	{ 
		tupleAck ++;
	}
		
	//METRIETAI TO SYNOLIKO ELAPSED TIME / THROUGHPUT APO THN STIGMH EKKINHSHS TOY BOLT MEXRI TH STIGMH POU STELNETAI TO METRIC STO CONSUMER (TIME BUCKET)
	public Object getValueAndReset() 
	{
		currentTime=DateTime.now();
		long secondsElapsed=(currentTime.getMillis() - startTime.getMillis())/1000;
		long throughput= this.tuplesProcessed / secondsElapsed;
		
	   /* String s= "Start: " + this.startTime.toString() + "  /  " + "Now: " + currentTime.toString() +  "  /  " + 
	    "Seconds Elapsed: " + secondsElapsed +  "  /  "  + " Tuples Processed: " + this.tuplesProcessed 
	    		+  "  /  "  + " Throughput: " + throughput + " tuples / second " + "(" + " Tuples Acked: " + this.tupleAck + ")";*/
	    
	    String s= "Seconds Elapsed: " + secondsElapsed +  "  /  "  + " Tuples Processed: " + this.tuplesProcessed 
	    		+  "  /  "  + " Throughput: " + throughput + " tuples / second " ;
	    		//+ "(" + " Tuples Acked: " + this.tupleAck + ")";
	    
		return s;
	}

}
